package cn.deschen.search.mysql.listeners;

import cn.deschen.biz.service.InformationSchemaService;
import cn.deschen.search.constants.TableInfo;
import cn.deschen.search.mysql.BinlogRowData;
import cn.deschen.search.mysql.senders.IndexSender;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.io.Serializable;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @Author hanbin_chen
 * @Description mysql binlog监听器
 * @DateTime 2021/6/9
 * @Version V1.0.0
 */
@Slf4j
public class BinlogListener implements BinaryLogClient.EventListener, BinaryLogClient.LifecycleListener{

    /**
     * 保存需要创建索引的表信息，用于其他表排除索引的作用
     */
    public static final List<String> needIndexTableList = new ArrayList<>();
    /**
     * binlog日志中的table_id与tableName的映射
     * 格式：table_id -> database.tableName
     */
    public static final Map<Long, String> tableInfoMap = new ConcurrentHashMap<>(16);

    public static final String SEPARATOR = ":";

    private InformationSchemaService schemaService;

    private IndexSender indexSender;

    static {
        for (TableInfo tableInfo : TableInfo.values()) {
            needIndexTableList.add(tableInfo.getDatasource() + SEPARATOR + tableInfo.getTableName());
        }
    }
    public BinlogListener(InformationSchemaService schemaService, IndexSender indexSender) {
        this.schemaService = schemaService;
        this.indexSender = indexSender;
    }


    /**
     * 监听binlog事件
     * binlog主要监听内容
     * Query	    1	7364	BEGIN                             sql语句开始执行，事务提交
     * Table_map	1	7442	table_id: 109 (数据库.数据表)       执行的数据库。数据表
     * Write_rows	1	7517	table_id: 109 flags: STMT_END_F   执行sql语句操作，这里是写操作
     * Xid	1	7548	COMMIT \* xid=834 *\                      提交sql语句，提交事务
     * @param event
     */
    @Override
    public void onEvent(Event event) {
        EventType type = event.getHeader().getEventType();
//        log.debug("LISTEN MYSQL EVENT: {}", event);

        /**
         * 获取监听的数据表
         */
        if (type == EventType.TABLE_MAP) {
            TableMapEventData data = event.getData();
            long tableId = data.getTableId();
            String tableName = data.getTable();
            String database = data.getDatabase();
            tableInfoMap.put(tableId, database + SEPARATOR + tableName);
            return;
        }

        /**
         * 跳过非数据库操作
         */
        if (!EventType.isWrite(type) && !EventType.isUpdate(type) &&!EventType.isDelete(type)) {
            return;
        }
        /**
         * 获取数据表信息
         */
        Long tableId = getTableId(event, type);
        String tableInfo = tableInfoMap.get(tableId);

        /**
         * 判断是否要对接下来的数据进行索引保存、更新、删除
         */
        if (!needIndexTableList.contains(tableInfo)) {
            return;
        }

        /**
         * 封装数据
         */
        String[] tableInfos = tableInfo.split(SEPARATOR);
        String database = tableInfos[0];
        String tableName = tableInfos[1];
        BinlogRowData rowData = buildRowData(database, tableName, type, event.getData());

//        log.debug("LISTEN MYSQL ROWDATA: {}", rowData);
        /**
         * 索引处理
         */
        indexSender.sender(rowData);
    }

    /**
     * 封装binlog日志数据
     *
     * @param database  数据库名
     * @param tableName 数据表名
     * @param eventType binlog操作类型
     * @param eventData binlog变更记录
     * @return
     */
    private BinlogRowData buildRowData(String database, String tableName, EventType eventType, EventData eventData) {
        BinlogRowData rowData = new BinlogRowData();
        rowData.setDatabase(database);
        rowData.setTableName(tableName);
        rowData.setEventType(eventType);
        rowData.setRowDataList(parseEventData2List(database, tableName, eventType, eventData));
        return rowData;
    }

    /**
     * 解析转换binlog变更记录
     * 格式：集合：字段名 -> 字段值
     *
     * @param database  数据库名
     * @param tableName 数据表名
     * @param eventType binlog操作类型
     * @param eventData binlog变更记录
     * @return
     */
    private List<Map<String, Serializable>> parseEventData2List(String database, String tableName, EventType eventType, EventData eventData) {
        List<Map<String, Serializable>> rowDataList = new ArrayList<>();
        for (Serializable[] rowDataValue : getRowDataValues(eventType, eventData)) {
            Map<String, Serializable> rowDataMap = new HashMap<>(16);
            int length = rowDataValue.length;

            for (int index = 0; index < length; ++index) {
                String colName = schemaService.listColumn(database, tableName, index + 1);
                // 如果没有则说明不关心这个列
                if (StringUtils.isEmpty(colName)) {
                    log.debug("ignore position: {}", index);
                    continue;
                }
                Serializable colValue = rowDataValue[index];
                rowDataMap.put(colName, colValue);
            }
            rowDataList.add(rowDataMap);
        }
        return rowDataList;
    }


    /**
     * 获取binlog变更记录的值 </br>
     * binlog变更记录的值是不带有字段映射的，只有字段所在索引位置与字段值的映射
     * @param eventType binlog操作类型
     * @param eventData binlog变更记录
     * @return
     */
    private List<Serializable[]> getRowDataValues(EventType eventType, EventData eventData) {

        if (EventType.isWrite(eventType)) {
            return ((WriteRowsEventData) eventData).getRows();
        }

        if (EventType.isUpdate(eventType)) {
            return ((UpdateRowsEventData) eventData).getRows().stream()
                    .map(Map.Entry::getValue)
                    .collect(Collectors.toList());
        }

        if (EventType.isDelete(eventType)) {
            return ((DeleteRowsEventData) eventData).getRows();
        }

        return Collections.emptyList();
    }

    /**
     * 获取数据表对应的唯一标识tableId
     */
    private Long getTableId(Event event, EventType type) {

        Long tableId = 0L;
        /**
         * 监听写操作
         */
        if (EventType.isWrite(type)) {
            WriteRowsEventData data = event.getData();
            tableId = data.getTableId();
        }
        /**
         * 监听更新操作
         */
        if (EventType.isUpdate(type)) {
            UpdateRowsEventData data = event.getData();
            tableId = data.getTableId();
        }

        /**
         * 监听更新操作
         */
        if (EventType.isDelete(type)) {
            DeleteRowsEventData data = event.getData();
            tableId = data.getTableId();
        }
        return tableId;
    }


    @Override
    public void onConnect(BinaryLogClient client) {
        log.debug("mysql binlog success connect");
    }

    @Override
    public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
        log.error("mysql binlog failure connect");
    }

    @Override
    public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
        log.error("mysql binlog failure deserialization");
    }

    @Override
    public void onDisconnect(BinaryLogClient client) {
        log.error("mysql binlog disconnect");
    }
}
